package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.TradeOrderBean;
import com.atguigu.edu.realtime.bean.TradeProvinceBean;
import com.atguigu.edu.realtime.bean.TradeSourceBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class DwsTradeSource {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        //TODO 2.检查点相关设置

        //TODO 3.从kafka主题中读取下单数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_source";
        String groupId = "dwd_trade_order_source";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //TODO 4.对读取数据进行类型转换    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        //TODO 5.按照订单省份paymenttype进行分组
        KeyedStream<JSONObject, String> orderaoymentTypeKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("payment_type"));
        //TODO 6.将流中数据类型进行转换    jsonObj - >实体类对象
        SingleOutputStreamOperator<TradeSourceBean> orderBeanDS = orderaoymentTypeKeyedDS.map(
                new MapFunction<JSONObject, TradeSourceBean>() {
                    @Override
                    public TradeSourceBean map(JSONObject jsonObj) throws Exception {
                        String orderId = jsonObj.getString("order_id");
                        String userId = jsonObj.getString("user_id");
                        String paymentId = jsonObj.getString("payment_id");
                        Double totalamount = jsonObj.getDouble("total_amount");
                        String paymenttype = jsonObj.getString("payment_type");
                        Long ts = jsonObj.getLong("create_time");
                        TradeSourceBean tradeSourceWindow = TradeSourceBean.builder()
                                .userId(userId)
                                .orderIdSet(new HashSet<String>(
                                        Collections.singleton(orderId)
                                ))
                                .paymentId(paymentId)
                                .paymenttype(paymenttype)
                                .totalAmount(totalamount)
                                .ts(ts)
                                .orderCount(0L)
                                .build();
                        return tradeSourceWindow;

                    }
                }
        );
        //orderBeanDS.print(">>>>");
        //TODO 7.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSourceBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSourceBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSourceBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSourceBean orderBean, long recordTimestamp) {
                                        return orderBean.getTs();
                                    }
                                }
                        )
        );
        //withWatermarkDS.print(">>>>");

        //TODO 9.按照支付来源维度进行分组
        KeyedStream<TradeSourceBean, String> paymentTypeKeyedDS = withWatermarkDS.keyBy(TradeSourceBean::getPaymenttype);
        //TODO 10.开窗
        WindowedStream<TradeSourceBean, String, TimeWindow> windowDS = paymentTypeKeyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 11.聚合
        SingleOutputStreamOperator<TradeSourceBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSourceBean>() {
                    @Override
                    public TradeSourceBean reduce(TradeSourceBean value1, TradeSourceBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.setTotalAmount(value1.getTotalAmount() + value2.getTotalAmount());
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        return value1;
                    }
                },
                new WindowFunction<TradeSourceBean, TradeSourceBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSourceBean> input, Collector<TradeSourceBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());

                        for (TradeSourceBean orderBean : input) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setTs(System.currentTimeMillis());
                            orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                            out.collect(orderBean);
                        }
                    }
                }
        );
        reduceDS.print(">>>>");
//        // TODO 10. 写出数据库
//        SinkFunction<TradeProvinceBean> jdbcSink = MyClickhouseUtil.<TradeProvinceBean>getSinkFunction(
//                "insert into dws_trade_order_province values(?,?,?,?,?,?)"
//        );
//        reduceDS.<TradeOrderBean>addSink(jdbcSink);
//        env.execute();
    }
}
